package pipeline

// 流水线处理器
type Process interface {
	// build
	Build(pipeCtx *PipeContext, pipeFun PipeFunc)
	// start
	Start()
	// add pipeline
	AddPipeLine(...PipeFunc)
	// get parent
	Parent() Process
	// list Children
	Children() []Process
	Result() <-chan Pipeline
}

// init 处理器
func NewProcessor(name string) Process {
	return &processor{
		Name: name,
	}
}

// 实际处理器
type processor struct {
	Id          string
	Name        string
	children    []Process
	parent      Process
	PipeFuncs   []PipeFunc
	PipeContext *PipeContext
}

func (p *processor) Parent() Process {
	return p.parent
}

func (p *processor) Result() <-chan Pipeline {
	return p.PipeContext.out
}

func (p *processor) Children() []Process {
	return p.children
}

func (p *processor) Build(pipeCtx *PipeContext, pipeFun PipeFunc) {
	p.PipeContext = pipeCtx
	p.PipeContext.out = p.build(pipeFun, pipeCtx.Pipelines...)
}

func (p *processor) Start() {
	out := make(<-chan Pipeline)
	out = p.PipeContext.out
	for _, pipeFunc := range p.PipeFuncs {
		out = p.process(out, pipeFunc)
	}
	p.PipeContext.out = out
}

func (p *processor) process(in <-chan Pipeline, pipeFun PipeFunc) <-chan Pipeline {
	out := make(chan Pipeline)
	go func() {
		defer close(out)
		for c := range in {
			out <- pipeFun(c)
		}
	}()
	return out
}

func (p *processor) build(pipeFun PipeFunc, pipelines ...Pipeline) <-chan Pipeline {
	out := make(chan Pipeline)
	go func() {
		defer close(out)
		for i := 0; i < len(pipelines); i++ {
			out <- pipeFun(pipelines[i])
		}
	}()
	return out
}

func (p *processor) AddPipeLine(pipeFunc ...PipeFunc) {
	p.PipeFuncs = append(p.PipeFuncs, pipeFunc...)
}
